package site.wanjiahao.factory;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.DefaultPromise;
import io.netty.util.concurrent.Promise;
import lombok.extern.slf4j.Slf4j;
import site.wanjiahao.handler.client.MyChannelDuplexHandler;
import site.wanjiahao.handler.client.PongMessageHandler;
import site.wanjiahao.handler.client.RpcResponseMessageHandler;
import site.wanjiahao.message.request.RpcRequestMessage;
import site.wanjiahao.protocol.BasedFrameDecoder;
import site.wanjiahao.protocol.MessageCodec;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 创建代理类，代理类的实际职责是使用客户端通道对象，发送调用方法的元数据信息
 * 给服务端，服务端调用实际的方法，封装返回结果并返回
 */
@Slf4j
public class ProxyFactory implements InvocationHandler {

    /**
     * 原生class类
     */
    private Class<?> primitiveClazz;

    /**
     * 客户端channel对象
     */
    private static Channel channel;

    /**
     * 异步获取结果
     */
    private static final Map<Long, Promise<Object>> futures = new ConcurrentHashMap<>();

    static {
        // 静态代码块，保证channel的单例
        initChannel();
        assert channel != null;
        log.debug("channel初始化完毕: {}", channel);
    }

    public static Channel getChannel() {
        return channel;
    }

    public static Promise<Object> getPromise(long key) {
        return futures.remove(key);
    }

    public <T> T getProxy(Class<T> t) {
        this.primitiveClazz = t;
        Class<?>[] interfaces = new Class[]{t};
        Object proxyInstance = Proxy.newProxyInstance(t.getClassLoader(), interfaces, this);
        return t.cast(proxyInstance);
    }


    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        RpcRequestMessage rpcRequestMessage = new RpcRequestMessage(
                primitiveClazz,
                method.getName(),
                method.getReturnType(),
                method.getParameterTypes(),
                args
        );
        // 放入消息对应的唯一promise，异步接受结果 channel.eventLoop()，异步获取结果使用的线程
        DefaultPromise<Object> promise = new DefaultPromise<>(channel.eventLoop());
        futures.put(rpcRequestMessage.getSequenceId(), promise);
        // 等待获取异步结果
        channel.writeAndFlush(rpcRequestMessage);
        // 同步等待结果, 等到结果不抛出异常(sync, get 抛出异常，await不抛出)
        return promise.get();
    }

    private static void initChannel() {
        LoggingHandler loggingHandler = new LoggingHandler(LogLevel.DEBUG);
        MessageCodec messageCodec = new MessageCodec();
        IdleStateHandler idleStateHandler = new IdleStateHandler(0, 3, 0);
        MyChannelDuplexHandler myChannelDuplexHandler = new MyChannelDuplexHandler();
        PongMessageHandler pongMessageHandler = new PongMessageHandler();
        RpcResponseMessageHandler rpcResponseMessageHandler = new RpcResponseMessageHandler();

        NioEventLoopGroup eventExecutors = new NioEventLoopGroup();

        ChannelFuture connectFuture = new Bootstrap()
                .group(eventExecutors)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) {
                        ch.pipeline()
                                .addLast(loggingHandler)
                                .addLast(new BasedFrameDecoder())
                                .addLast(messageCodec)
                                .addLast(idleStateHandler)
                                .addLast(myChannelDuplexHandler)
                                .addLast(pongMessageHandler)
                                .addLast(rpcResponseMessageHandler);
                    }
                })
                .connect("127.0.0.1", 8080);

        try {
            channel = connectFuture.sync().channel();
            // 异步关闭线程
            channel.closeFuture().addListener((ChannelFutureListener) future -> {
                // 结束进程
                eventExecutors.shutdownGracefully();
            });
        } catch (InterruptedException interruptedException) {
            interruptedException.printStackTrace();
        }
    }

}
